package com.asap.demo.function;

import com.asap.demo.model.BeanField;
import com.asap.demo.rete.ReteDemo4;
import com.asap.demo.utils.DateUtil;
import com.asap.demo.utils.Utils;
import com.asap.rule.StandardEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @author wangbh
 * @Description: es计算接口
 * @date 2021/11/26 17:26
 */
public class dealMapFunctionForEs implements MapFunction<StandardEvent, StandardEvent> {
    private static final Logger logger = LoggerFactory.getLogger(dealMapFunctionForEs.class);
    private SimpleDateFormat simpleDate = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
    private String indexCreateMode="DAY";
    @Override
    public StandardEvent map(StandardEvent value) throws Exception {
        StandardEvent standardEvent = value;
        if(standardEvent.getAllFields().isEmpty()){
            return null;
        }
        String indexName = null;
        int count=0;
        String collectTime = DateUtil.dateInfo();
        if (standardEvent.getAllFields().containsKey("CREATE_TIME")) {
            collectTime = (String) standardEvent.getField("CREATE_TIME");
        }
        standardEvent.getAllFields().put("collectTime", collectTime);
        standardEvent.getAllFields().put("recordTime", simpleDate.format(new Date()));
        standardEvent.getAllFields().put("rawMsg",value);
        String collection = "maxs_standard";
        if (standardEvent.getAllFields().containsKey("collectionName")) {
            collection = (String) standardEvent.getField("collectionName");
        }

        Date date= DateUtil.parseDate(collectTime,DateUtil.DATE_FORMAT_FULL);
        String time = DateUtil.parseDateToString(date,DateUtil.DATE_FORMAT);
        if(logger.isDebugEnabled()){
            logger.debug(Utils.join("","collectTime:",collectTime,"time:",time));

        }

        if("DAY".equals(indexCreateMode)){
            indexName = Utils.join("_",collection,time.substring(0,8));
        } else if("MONTH".equals(indexCreateMode)){
            indexName = Utils.join("_",collection,time.substring(0,6));
        } else if("HOUR".equals(indexCreateMode)){
            indexName = Utils.join("_",collection,time.substring(0,10));
        } else {
            indexName = collection;
        }
        if (indexName.startsWith("-") || indexName.startsWith("_") || indexName.startsWith("+"))
        {
            count++;
            if (count < 2)
            {
                logger.error("",null,Utils.join("","当前设置的索引名称 ",
                        collection," 初始化后名称 ",indexName," 不合法，其初始化索引名称变更为 unknown "));
            }
            indexName = "unknown";
        }
        standardEvent.getAllFields().put("indexName", indexName);

        return standardEvent;
    }
}
